package demo.components.kafka;

import java.io.IOException;
import java.util.HashMap;
import java.util.Map;
import java.util.Properties;

import org.apache.commons.lang.StringUtils;
import org.apache.kafka.clients.producer.KafkaProducer;
import org.apache.kafka.clients.producer.ProducerConfig;
import org.apache.kafka.clients.producer.ProducerRecord;
import org.apache.kafka.common.serialization.StringSerializer;
import org.apache.log4j.Logger;


public class KafkaClient {
    private static final Logger logger = Logger.getLogger(KafkaClient.class);
    private static KafkaClient kafkaClient = new KafkaClient();
    private Map<String, MsgHandler> hadlerMap;
    private static final int DEFAULT_MB_BUFFER = 32;
    private static final int DEFAULT_MS_LINGER = 5;
    private static final int SLOW_KAFKA_SEND = 10;

    public static void send(String message, String handlerKey) throws Exception {
        send(null, message, handlerKey);
    }

    public static void send(String key, String message, String handlerKey) throws Exception {
        logger.debug("raw:" + message);
        kafkaClient.sendMsg(key, message, handlerKey);
    }

    public static void initOrReset(String key, KafkaConfig kafkaConfig) throws IOException {
        kafkaClient.destroy();
        kafkaClient.init(key, kafkaConfig);
    }

    public void init(String key, KafkaConfig kafkaConfig) throws IOException {
        if (hadlerMap == null) {
            hadlerMap = new HashMap<String, MsgHandler>();
        }
        hadlerMap.put(key, new MsgHandler(kafkaConfig.getTopic(), kafkaConfig.getClientId(), kafkaConfig.getBrokers(),
                kafkaConfig.getBatchSize(), kafkaConfig.getRetry(), kafkaConfig.isAck()));
    }

    public void destroy() {
        if (null != hadlerMap && hadlerMap.size() != 0) {
            for (Map.Entry<String, MsgHandler> entry : hadlerMap.entrySet()) {
                entry.getValue().close();
            }
        }
    }

    public void sendMsg(String key, String message, String handlerKey) throws Exception {
        hadlerMap.get(handlerKey).sendMsg(key, message);
    }

    private class MsgHandler {
        private final String topicName;
        private final KafkaProducer<String, String> producer;

        public MsgHandler(String topicName, String clientId, String brokers, int batchSize, int retry, boolean ack) {
            this.topicName = topicName;

            Properties props = new Properties();
            props.put(ProducerConfig.BOOTSTRAP_SERVERS_CONFIG, brokers);
            props.put(ProducerConfig.VALUE_SERIALIZER_CLASS_CONFIG, StringSerializer.class.getName());
            props.put(ProducerConfig.KEY_SERIALIZER_CLASS_CONFIG, StringSerializer.class.getName());
            props.put(ProducerConfig.CLIENT_ID_CONFIG, clientId);
            if (ack) {
                // 配置可以设定发送消息后是否需要Broker端返回确认,0（不需要确认）,1（仅需要leadre确认）,all（
                // 需要leader和ISR）
                props.put(ProducerConfig.ACKS_CONFIG, "1");
                props.put(ProducerConfig.RETRIES_CONFIG, retry);
            } else {
                props.put(ProducerConfig.ACKS_CONFIG, "0");
            }
            props.put(ProducerConfig.BATCH_SIZE_CONFIG, batchSize * CommonConst.KB);
            props.put(ProducerConfig.LINGER_MS_CONFIG, DEFAULT_MS_LINGER);
            props.put(ProducerConfig.BUFFER_MEMORY_CONFIG, DEFAULT_MB_BUFFER * CommonConst.KB * CommonConst.KB);

            producer = new KafkaProducer<String, String>(props);
        }

        public void close() {
            producer.close();
        }

        public void sendMsg(String key, String message) throws Exception {
            long start = System.currentTimeMillis();
            if (StringUtils.isNotBlank(key)) {
                producer.send(new ProducerRecord<String, String>(this.topicName, key, message));
            } else {
                producer.send(new ProducerRecord<String, String>(this.topicName, message));
            }

            long time = System.currentTimeMillis() - start;
            if (time > SLOW_KAFKA_SEND) {
                logger.info("kafka send cost" + time);
            }
        }
    }

}
